package com.felix.thread.producer.consumer;

import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Producer
 *
 * @author xuejinhua
 * @date 2018/10/25 16:06
 */
public class ProducerConsumerBlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> myQueue = new LinkedBlockingQueue<>();
        ExecutorService service = Executors.newCachedThreadPool();
        BlockingProducer p1 = new BlockingProducer(myQueue, "p1");
        BlockingProducer p2 = new BlockingProducer(myQueue, "p2");
        BlockingProducer p3 = new BlockingProducer(myQueue, "p3");
        BlockingProducer p4 = new BlockingProducer(myQueue, "p4");
        BlockingConsumer c1 = new BlockingConsumer(myQueue, "c1");
        BlockingConsumer c2 = new BlockingConsumer(myQueue, "c2");
        BlockingConsumer c3 = new BlockingConsumer(myQueue, "c3");
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(p4);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
        // 3.睡一会儿然后尝试停止生产者
        Thread.sleep(10 * 1000);
        p1.stop();
        p2.stop();
        p3.stop();
        p4.stop();
        // 4.再睡一会儿关闭线程池
        Thread.sleep(3000);
        service.shutdown();
        System.exit(0);
    }

}

class BlockingProducer implements Runnable {

    private BlockingQueue<Integer> queue;
    private String name;
    private volatile boolean isRunning = true;
    /**
     * 总数，原子操作
     */
    private static AtomicInteger count = new AtomicInteger();

    public BlockingProducer(BlockingQueue<Integer> queue, String name) {
        this.queue = queue;
        this.name = name;
    }

    @Override
    public void run() {
        int data;
        Random r = new Random();

        System.out.println("start producer id = " + name);
        try {
            while (isRunning) {
                // 模拟延迟
                Thread.sleep(r.nextInt(1000));
                // 往阻塞队列中添加数据
                // 构造任务数据
                data = count.incrementAndGet();
                System.out.println("producer " + name + " create data：" + data + ", size：" + queue.size());
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.err.println("failed to put data：" + data);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.interrupted();
        }

    }

    protected void stop() {
        isRunning = false;
        System.out.println("Producer " + name + " stop");
    }
}


class BlockingConsumer implements Runnable {
    private BlockingQueue<Integer> queue;

    private String name;

    public BlockingConsumer(BlockingQueue<Integer> theQueue, String name) {
        this.queue = theQueue;
        this.name = name;
    }

    @Override
    public void run() {
        int data;
        Random r = new Random();

        System.out.println("start consumer id = " + name);
        try {
            while (true) {
                // 模拟延迟
                Thread.sleep(r.nextInt(1000));
                // 从阻塞队列中获取数据
                if (!queue.isEmpty()) {
                    data = queue.take();
                    System.out.println("consumer " + name + " consume data：" + data + ", size：" + queue.size());
                } else {
                    System.out.println("Queue is empty, consumer " + name + " is waiting, size：" + queue.size());
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }

    }
}
